package com.coolingme.kafka.producer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.coolingme.dto.KafkaDataDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Kafka生产者，整合spring
 *
 * @author wangyue
 * @date 2021/10/25 11:20
 */
@Slf4j
@RestController
@RequestMapping("kafka")
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * topic名称
     */
    private String TOPIC_NAME = "test-topic";

    /**
     * 发送用户消息
     */
    @GetMapping("/send/{msg}")
    public JSONObject sendMessage(@PathVariable String msg) {

        JSONObject jsonObject = new JSONObject();
        jsonObject.put("id", RandomUtil.randomLong(999L));
        jsonObject.put("time", DateUtil.now());
        jsonObject.put("msg", msg);

        //使用kafka模板发送消息
        kafkaTemplate.send(TOPIC_NAME, jsonObject.toJSONString()).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            //发送失败
            @Override
            public void onFailure(Throwable throwable) {
                log.error("发送失败->" + jsonObject.toJSONString());
                log.error(throwable.getMessage());
            }

            //发送成功
            @Override
            public void onSuccess(SendResult<String, String> sendResult) {
                RecordMetadata metadata = sendResult.getRecordMetadata();
                log.info("发送成功->" + jsonObject.toJSONString() + " | " + metadata.partition() + "-" + metadata.offset());
            }
        });
        return jsonObject;
    }

    /**
     * 发送对象
     */
    @GetMapping("/sendObj/{msg}")
    public KafkaDataDTO sendObj(@PathVariable String msg) {

        KafkaDataDTO kafkaDataDTO = new KafkaDataDTO(RandomUtil.randomLong(999L), DateUtil.date(), msg);
        String dtoJson = JSON.toJSONString(kafkaDataDTO);
        String topicName = "coolingme";

        //使用kafka模板发送消息
        kafkaTemplate.send(topicName, dtoJson).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            //发送失败
            @Override
            public void onFailure(Throwable throwable) {
                log.error("发送失败->" + dtoJson);
                log.error(throwable.getMessage());
            }

            //发送成功
            @Override
            public void onSuccess(SendResult<String, String> sendResult) {
                RecordMetadata metadata = sendResult.getRecordMetadata();
                log.info("发送成功->" + dtoJson + " | " + metadata.partition() + "-" + metadata.offset());
            }
        });
        return kafkaDataDTO;
    }

    public AtomicInteger integer = new AtomicInteger(0);

    /**
     * 同步发送消息
     *
     * @param msg 消息
     * @return java.lang.String 消息体
     * @author wangyue
     * @date 2023/12/7 15:36
     */
    @GetMapping("/sendSync/{msg}")
    public JSONObject sendMessageSync(@PathVariable String msg) {
        String topic = TOPIC_NAME;

        KafkaDataDTO kafkaDataDTO = new KafkaDataDTO(RandomUtil.randomLong(999L), DateUtil.date(), msg + integer.incrementAndGet());
        String dtoJson = JSON.toJSONString(kafkaDataDTO);
        try {
            sendMessage(topic, dtoJson);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return JSONObject.parseObject(dtoJson);
    }

    public Boolean sendMessage(String topic, String message) throws Exception {

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);

        // 同步等待返回，设置5秒超时
        SendResult<String, String> sender = this.kafkaTemplate.send(record).get(5L, TimeUnit.SECONDS);
        if (sender != null) {
            //有数据代表发送成功
            log.info("发送成功-> {} ", message);
            return true;
        }
        log.error("发送失败-> {} ", message);
        return false;
    }


}
